package operator.combine;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import util.Printer;
import util.RxJavaHelper;
import util.ThreadHelper;

class MergeOperator {

    public static void main(String[] args) {
        merge();
//        mergeArray();
//        mergeArrayDelayError();
    }

    static void merge() {
//        Observable.merge(
//                Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, String>() {
//                    @Override
//                    public String apply(Long aLong) throws Exception {
//                        return "A" + aLong;
//                    }
//                }),
//                Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, String>() {
//                    @Override
//                    public String apply(Long aLong) throws Exception {
//                        return "B" + aLong;
//                    }
//                }))
//                .subscribe(getConsumer());

        Observable.concat(
                Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Exception {
                        return "A" + aLong;
                    }
                }),
                Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Exception {
                        return "B" + aLong;
                    }
                }))
                .subscribe(RxJavaHelper.getConsumer());

        ThreadHelper.sleep();

    }

    static void mergeArray() {
        Observable.mergeArray(
                Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Exception {
                        return "A" + aLong;
                    }
                }),
                Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Exception {
                        return "B" + aLong;
                    }
                }))
                .subscribe(RxJavaHelper.getConsumer());
    }





    static void mergeArrayDelayError(){
        Observable.mergeArrayDelayError(RxJavaHelper.getErrorObservable(), Observable.just("123"))
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Printer.print("mergeArrayDelayError s = " + s);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Printer.print("mergeArrayDelayError throwable = " + throwable);
                    }
                });
    }

}
